package club.vann.kafka.mock.service;

import club.vann.kafka.mock.kafka.ConstantTopic;
import club.vann.kafka.mock.kafka.KafkaConsumer;
import club.vann.kafka.mock.kafka.KafkaProducer;
import com.cyberx.lunyu.AcceptorOuterClass;
import com.google.protobuf.InvalidProtocolBufferException;
import java.util.UUID;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

@Slf4j
@Service
public class AcceptorOutputService {
  private final KafkaProducer kafkaProducer;
  private final QuoteAcceptorService quoteAcceptorService;

  public AcceptorOutputService(
      KafkaConsumer kafkaConsumer,
      KafkaProducer kafkaProducer,
      QuoteAcceptorService quoteAcceptorService) {
    this.kafkaProducer = kafkaProducer;
    this.quoteAcceptorService = quoteAcceptorService;
    kafkaConsumer.subTopic(ConstantTopic.TOPIC_ACCEPTOR_OUTPUT, "", this::sub);
  }

  private void defaultRequest(AcceptorOuterClass.Acceptor.AcceptorRequest request) {
    String clientRequestId = request.getClientRequestId();
    String responseTopic = request.getTopic();
    if (!ObjectUtils.isEmpty(responseTopic)) {
      AcceptorOuterClass.Acceptor.AcceptorResponse acceptorResponse =
          of(clientRequestId, AcceptorOuterClass.Acceptor.Status.COMPLETED);
      this.kafkaProducer.send(responseTopic, clientRequestId, acceptorResponse.toByteArray());
    }
  }

  private AcceptorOuterClass.Acceptor.AcceptorResponse of(
      String clientRequestId, AcceptorOuterClass.Acceptor.Status status) {
    return AcceptorOuterClass.Acceptor.AcceptorResponse.newBuilder()
        .setRequestId(UUID.randomUUID().toString())
        .setClientRequestId(clientRequestId)
        .setStatus(status)
        .putData("MESS", "执行成功")
        .build();
  }

  private void sub(byte[] bytes) {
    try {
      AcceptorOuterClass.Acceptor.AcceptorRequest acceptorRequest =
          AcceptorOuterClass.Acceptor.AcceptorRequest.parseFrom(bytes);
      AcceptorOuterClass.Acceptor.RequestType requestType = acceptorRequest.getType();
      switch (requestType) {
        case QUOTE_REQUEST -> quoteAcceptorService.request(acceptorRequest);
        case QUOTE_CONFIRM -> quoteAcceptorService.confirm(acceptorRequest);
        case QUOTE_CANCEL -> quoteAcceptorService.cancel(acceptorRequest);
        default -> defaultRequest(acceptorRequest);
      }
    } catch (InvalidProtocolBufferException e) {
      log.error(e.getMessage());
    }
  }
}
